package com.example.cloud_disk.hdfsApi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Files;

/**
 * @author majun
 * @package com.example.cloud_disk.hdfsApi
 * @date 2024/10/22 20:12
 */
@Service
public class HdfsOperationsService {

    @Value("${hadoop.fs.defaultFS}")
    private String hdfsUri;

    /**
     * 创建文件夹
     *
     * @param dirPath
     */
    public String createDirectory(String dirPath) throws IOException, InterruptedException, URISyntaxException {
        // 1.创建配置文件
        Configuration conf = new Configuration();
        // 2.获取文件系统，HDFS文件系统
        FileSystem fs = FileSystem.get(new URI(hdfsUri), conf, "root");
        // 3.调用API操作，创建文件夹
        fs.mkdirs(new Path(dirPath));
        // 4.设置文件夹权限为777（rwxrwxrwx）
        FsPermission permission = new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL);
        fs.setPermission(new Path(dirPath), permission);
        // 5.关闭资源
        fs.close();

        return "文件夹创建成功，并设置权限为777";
    }

    /**
     * 文件向HDFS上传
     *
     * @param file
     * @param hdfsPath
     * @return
     */
    public String uploadingFiles(MultipartFile file, String hdfsPath) throws InterruptedException, IOException, URISyntaxException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(new URI(hdfsUri), conf, "root");

        // 创建临时文件
        java.nio.file.Path tempPath = Files.createTempFile("tempFile", ".tmp");
        try {
            file.transferTo(tempPath.toFile());

            // 创建输入输出流
            try (InputStream in = new FileInputStream(tempPath.toFile());
                 OutputStream out = fs.create(new Path(hdfsPath), true)) {
                IOUtils.copyBytes(in, out, 4096, true);
            }

            return "文件上传成功";
        } finally {
            Files.deleteIfExists(tempPath); // 清理临时文件
        }
    }

    /**
     * 用户文件下载
     *
     * @param hdfsPath HDFS文件路径
     * @param fileName 文件名称
     * @return
     * @throws URISyntaxException
     * @throws IOException
     */
    public String fileDownload(String hdfsPath, String fileName) throws URISyntaxException, IOException {

        FileSystem fs = FileSystem.get(new URI(hdfsUri), new Configuration());

        InputStream is = fs.open(new Path(hdfsPath));

        OutputStream out = new FileOutputStream(new File("D:/" + fileName));

        IOUtils.copyBytes(is, out, 4096, true);

        return "文件下载成功";

    }

    /**
     * 用户删除文件
     *
     * @param hdfsPath
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public String fileDelete(String hdfsPath) throws IOException, InterruptedException {

        Configuration conf = new Configuration();

        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf, "root");

        Boolean success = fs.delete(new Path(hdfsPath), true);

        return success.toString();

    }

    /**
     * 删除用户文件夹
     *
     * @param hdfsPath
     * @return
     * @throws IOException
     * @throws InterruptedException
     */
    public String deleteFolder(String hdfsPath) throws IOException, InterruptedException {
        Configuration conf = new Configuration();
        FileSystem fs = null;
        try {
            fs = FileSystem.get(URI.create(hdfsUri), conf, "root");
            Boolean success = fs.delete(new Path(hdfsPath), true);
            return success.toString();
        } finally {
            if (fs != null) {
                fs.close(); // 确保FileSystem资源被关闭
            }
        }
    }

}
